package com.yhy.common.rabbitmq.utils;

import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.yhy.common.rabbitmq.callback.ConfirmCallback;
import com.yhy.common.rabbitmq.callback.ReturnsCallback;
import com.yhy.common.rabbitmq.config.DefaultDelayConfig;
import com.yhy.common.rabbitmq.model.MessageData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;

import javax.annotation.PostConstruct;

@Slf4j
@Component
public class RabbitMqUtil {

    public static final String X_DELAYED_MESSAGE = DefaultDelayConfig.X_DELAYED_MESSAGE;
    static RabbitTemplate rabbitTemplate;
    @Autowired
    RabbitTemplate rabbitTemplateBean;
    @Autowired
    ConfirmCallback confirmCallback;
    @Autowired
    ReturnsCallback returnsCallback;

    /**
     * 发送普通消息
     *
     * @param exchange   exchange
     * @param routingKey routingKey
     * @param data       data
     * @return correlationDataId消息id
     */
    public static String sendMsg(String exchange, String routingKey, Object data) {
        String correlationDataId = IdUtil.simpleUUID();
        sendMsg(exchange, routingKey, data, correlationDataId);
        return correlationDataId;
    }

    /**
     * 发送普通消息
     *
     * @param exchange          exchange
     * @param routingKey        routingKey
     * @param data              data
     * @param correlationDataId correlationDataId
     */
    public static void sendMsg(String exchange, String routingKey, Object data, String correlationDataId) {
        checkQueueParam(exchange, routingKey, data, correlationDataId);
        rabbitTemplate.convertAndSend(exchange, routingKey, new MessageData<>(correlationDataId, data),
                message -> {
                    message.getMessageProperties().setMessageId(correlationDataId);
                    message.getMessageProperties().setCorrelationId(correlationDataId);
                    return message;
                }, new CorrelationData(correlationDataId));
    }

    /**
     * 发送延迟消息: 死信+TTL 实现
     *
     * @param exchange   exchange
     * @param routingKey routingKey
     * @param data       data
     * @param delayTime  延迟时间  单位/millisecond
     * @return correlationDataId消息id
     */
    public static String sendDelayMsg(String exchange, String routingKey, Object data, Long delayTime) {
        String correlationDataId = IdUtil.simpleUUID();
        sendDelayMsg(exchange, routingKey, data, correlationDataId, delayTime);
        return correlationDataId;
    }

    /**
     * 发送延迟消息: 死信+TTL 实现
     *
     * @param exchange          exchange
     * @param routingKey        routingKey
     * @param data              data
     * @param correlationDataId correlationDataId
     * @param delayTime         延迟时间  单位/millisecond
     */
    public static void sendDelayMsg(String exchange, String routingKey, Object data, String correlationDataId, Long delayTime) {
        checkDelayQueueParam(exchange, routingKey, data, correlationDataId, delayTime);
        rabbitTemplate.convertAndSend(exchange, routingKey, new MessageData<>(correlationDataId, data),
                message -> {
                    message.getMessageProperties().setExpiration(String.valueOf(delayTime));
                    message.getMessageProperties().setMessageId(correlationDataId);
                    message.getMessageProperties().setCorrelationId(correlationDataId);
                    return message;
                }, new CorrelationData(correlationDataId));

    }

    /**
     * 发送延迟消息: 借助rabbitMq插件rabbitmq-plugins发送延迟消息
     *
     * @param exchange   exchange
     * @param routingKey routingKey
     * @param data       data
     * @param delayTime  延迟时间  单位/millisecond 范围1-2的32次方-1
     * @return correlationDataId消息id
     */
    public static String sendDelayMsgByPlugin(String exchange, String routingKey, Object data, Long delayTime) {
        String correlationDataId = IdUtil.simpleUUID();
        sendDelayMsgByPlugin(exchange, routingKey, data, correlationDataId, delayTime);
        return correlationDataId;
    }

    /**
     * 发送延迟消息: 借助rabbitMq插件rabbitmq-plugins发送延迟消息
     * <p>
     * 借助rabbitMq插件rabbitmq-plugins enable rabbitmq_delayed_message_exchange实现
     * 消息发送回调机制spring支持的不友好，会触发returnsCallback机制, 这种场景最好手动处理该异常。
     * <p>
     * 现有实现是消息先发送到了exchange 到达过期时间再路由到对应queue
     *
     * @param exchange          exchange
     * @param routingKey        routingKey
     * @param data              data
     * @param correlationDataId correlationDataId
     * @param delayTime         延迟时间  单位/millisecond 范围1-2的32次方-1
     */
    public static void sendDelayMsgByPlugin(String exchange, String routingKey, Object data, String correlationDataId, Long delayTime) {
        checkDelayQueueParam(exchange, routingKey, data, correlationDataId, delayTime);
        rabbitTemplate.convertAndSend(exchange, routingKey, new MessageData<>(correlationDataId, data),
                message -> {
                    message.getMessageProperties().setHeader(DefaultDelayConfig.X_DELAY, delayTime);
                    message.getMessageProperties().setMessageId(correlationDataId);
                    message.getMessageProperties().setCorrelationId(correlationDataId);
                    message.getMessageProperties().setType(X_DELAYED_MESSAGE);
                    return message;
                },
                new CorrelationData(correlationDataId)
        );

    }

    private static void checkDelayQueueParam(String exchange, String routingKey, Object data, String correlationDataId, Long delayTime) {
        Assert.isTrue(delayTime >= 0, "延迟时间不合法");
        checkQueueParam(exchange, routingKey, data, correlationDataId);
    }

    private static void checkQueueParam(String exchange, String routingKey, Object data, String correlationDataId) {
        Assert.isTrue(StrUtil.isNotEmpty(exchange), "交换机名称为空");
        Assert.isTrue(StrUtil.isNotEmpty(routingKey), "routingKey为空");
        Assert.isTrue(StrUtil.isNotEmpty(correlationDataId), "消息ID为空");
        Assert.isTrue(ObjectUtil.isNotEmpty(data), "消息体不能为空");
    }

    /**
     * 用增强的方式显式增强, 解决实例化之后returnsCallback属性不生效
     */
    @PostConstruct
    public void init() {
        rabbitTemplate = rabbitTemplateBean;
        rabbitTemplate.setReturnsCallback(returnsCallback);
        rabbitTemplate.setConfirmCallback(confirmCallback);
    }

}
